package com.ywps.vaas.framework.dao

import com.ywps.vaas.framework.util.{EnvUtil, JdbcUtil, KafkaUtil, MysqlUtils}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.{LongType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, DataFrameReader, Row, SQLContext, SaveMode, SparkSession}
import org.slf4j.LoggerFactory

import java.util.Properties

/**
 * 数据持久层接口
 */
trait TDao {

  val logger = LoggerFactory.getLogger(this.getClass)
  val reader: DataFrameReader = EnvUtil.take().read

  /**
   * 填充自增主键id列
   * @param dataFrame
   * @param spark
   * @return
   */
  def addAutoIncreaceIdColumn(dataFrame: DataFrame,spark: SparkSession):DataFrame={
      addAutoIncreaceIdColumn(dataFrame, spark,"id")
  }

  /**
   * 填充自增主键列，主键名称自定义
   * @param dataFrame
   * @param spark
   * @param keyName 主键名称
   * @return
   */
  def addAutoIncreaceIdColumn(dataFrame: DataFrame,spark: SparkSession,keyName:String):DataFrame={
    val schema: StructType = dataFrame.schema.add(StructField(keyName, LongType))
    // DataFrame转RDD, 然后调⽤ zipWithIndex
    val dfRDD: RDD[(Row, Long)] = dataFrame.rdd.zipWithIndex()
    // 将id字段合并在⼀起，merge顺序不可修改，因为添加id的schema字段在最后⼀个
    val rowRDD: RDD[Row] = dfRDD.map(tp => Row.merge(tp._1,Row(tp._2)))
    spark.createDataFrame(rowRDD, schema)
  }

  def readTextFile(path: String) = {
    reader.textFile(path)
  }

  def saveTextFile(path: String): Unit = {
    logger.error("方法：{}，尚未实现", "saveTextFile")
  }

  def readJdbc(tableName: String): DataFrame = {
    reader.jdbc(JdbcUtil.getUrl(), tableName, JdbcUtil.getConf())
  }

  /**
   * 读取指定数据库中的表
   * @param tableName 表名称
   * @param databaseName 数据库名
   * @return
   */
  def readJdbc(tableName: String, databaseName:String): DataFrame = {
    if (databaseName.nonEmpty && !"".equals(databaseName.trim)){
      val url: String=JdbcUtil.getUrl()
      val i: Int = url.lastIndexOf("/")
      val j: Int = url.indexOf("?")
      val start: String = url.substring(0,i+1)
      val end: String = url.substring(j, url.length)
      val realUrl: String=start+databaseName+end
      reader.jdbc(realUrl, tableName, JdbcUtil.getConf())
    }else{
      logger.error("数据库名不能为空")
      readJdbc(tableName)
    }

  }

  def readJdbc(tableName: String, prop: Properties): DataFrame = {
    reader.jdbc(JdbcUtil.getUrl(), tableName, prop)
  }

  def readJdbc(url: String, tableName: String, prop: Properties): DataFrame = {
    reader.jdbc(url, tableName, prop)
  }

  def writeJdbc(dataFrame: DataFrame, model: SaveMode, tableName: String, url: String, prop: Properties) = {
    dataFrame.write.mode(model).jdbc(url, tableName, prop)
  }

  def writeJdbc(dataFrame: DataFrame, model: SaveMode, tableName: String): Unit = {
    dataFrame.write.mode(model).jdbc(JdbcUtil.getUrl(), tableName, JdbcUtil.getConf())
  }

  def writeJdbc(dataFrame: DataFrame, tableName: String): Unit = {
    dataFrame.write.mode(SaveMode.Append).jdbc(JdbcUtil.getUrl(), tableName, JdbcUtil.getConf())
  }

  /**
   * 将DataFrame所有类型(除id外)转换为String后,通过c3p0的连接池方法，向mysql写入数据
   *
   * @param tableName       表名
   * @param resultDateFrame DataFrame
   */
  def saveDFtoDBUsePool(tableName: String, resultDateFrame: DataFrame): Unit = {
    MysqlUtils.saveDFtoDBUsePool(tableName, resultDateFrame)
  }

  /**
   * 清空数据表
   *
   * @param sqlContext
   * @param mysqlTableName
   * @return
   */
  def truncateMysqlTable(sqlContext: SQLContext, mysqlTableName: String): Boolean = {
    MysqlUtils.truncateMysqlTable(sqlContext, mysqlTableName)
  }

  /**
   * 删除数据表
   *
   * @param sqlContext
   * @param mysqlTableName
   * @return
   */
  def dropMysqlTable(sqlContext: SQLContext, mysqlTableName: String): Boolean = {
    MysqlUtils.dropMysqlTable(sqlContext, mysqlTableName)
  }

  /**
   * 删除表中的数据
   *
   * @param sqlContext
   * @param mysqlTableName
   * @param condition
   * @return
   */
  def deleteMysqlTableData(mysqlTableName: String, condition: String): Boolean = {
    MysqlUtils.deleteMysqlTableData(mysqlTableName, condition)
  }

  /**
   * 保存DataFrame 到 MySQL中，如果表不存在的话，会自动创建
   *
   * @param tableName
   * @param resultDateFrame
   */
  def saveDFtoDBCreateTableIfNotExist(tableName: String, resultDateFrame: DataFrame,columnSize:Int=255): Unit = {
    MysqlUtils.saveDFtoDBCreateTableIfNotExist(tableName, resultDateFrame,columnSize)
  }

  /**
   * 保存DataFrame 到 MySQL中，如果表不存在的话，会自动创建;若已存在，则先删除再自动创建
   *
   * @param tableName
   * @param resultDateFrame
   */
  def saveDFtoDBOverrideTable(tableName: String, sqlContext: SQLContext, resultDataFrame: DataFrame,columnSize:Int=255,model:String="insert"): Unit = {
    val isExit: Boolean = MysqlUtils.checkTableExist(tableName)
    logger.info("目标mysql表：{}检查结果：{}",{tableName},{isExit})
    isExit match {
      case true => {
        MysqlUtils.dropMysqlTable(sqlContext, tableName)
        MysqlUtils.saveDFtoDBCreateTableIfNotExist(tableName, resultDataFrame,columnSize, model)
      }
      case false => MysqlUtils.saveDFtoDBCreateTableIfNotExist(tableName, resultDataFrame,columnSize, model)
    }

  }

  /**
   * 通过insertOrUpdate的方式把DataFrame写入到MySQL中，注意：此方式，必须对表设置主键
   *
   * @param tableName
   * @param resultDateFrame
   * @param updateColumns
   */
  def insertOrUpdateDFtoDBUsePool(tableName: String, resultDateFrame: DataFrame, updateColumns: Array[String]): Unit = {
    MysqlUtils.insertOrUpdateDFtoDBUsePool(tableName, resultDateFrame, updateColumns,null)
  }

  /**
   * 通过insertOrUpdate的方式把DataFrame写入到MySQL指定数据库的表中，注意：此方式，必须对表设置主键
   *
   * @param tableName
   * @param databaseName
   * @param resultDateFrame
   * @param updateColumns
   */
  def insertOrUpdateDFtoDBUsePool(tableName: String,databaseName: String, resultDateFrame: DataFrame, updateColumns: Array[String]): Unit = {
    MysqlUtils.insertOrUpdateDFtoDBUsePool(tableName, resultDateFrame, updateColumns,databaseName)
  }

  /**
   * 如果数据表不存在,根据DataFrame的字段创建数据表,数据表字段顺序和dataFrame对应
   * 若DateFrame出现名为id的字段,将其设为数据库主键(int,自增,主键),其他字段会根据DataFrame的DataType类型来自动映射到MySQL中
   *
   * @param tableName 表名
   * @param df        dataFrame
   * @return
   */
  def createTableIfNotExist(tableName: String, df: DataFrame): AnyVal = {
    MysqlUtils.createTableIfNotExist(tableName, df)
  }

  /**
   * 静态数据写入kafka
   * @param dataFrame 静态数据集
   * @param topic 主题
   */
  def writeDfToKafka(dataFrame: DataFrame,topic:String): Unit ={
    KafkaUtil.writeDfToKafka(dataFrame,topic)
  }

  /**
   * 数据流写入kafka
   * @param dataFrame 数据流
   * @param topic 主题
   */
  def writeStreamDfToKafka(dataFrame: DataFrame,topic:String): Unit ={
    KafkaUtil.writeStreamDfToKafka(dataFrame,topic)
  }
}
